package com.atguigu.market_analysis

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.{ProcessWindowFunction, WindowFunction}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 页面广告点击量统计
 * 页面广告按照省份划分的点击量的统计
 * 对一段时间内（比如一天内）的用户点击行为进行约束，如果对同一个广告点击超过一定限额（比如100次），应该把该用户加入黑名单并报警
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.market_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/13 18:02
 */
object AdStatisticsByGeo {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


//    env.enableCheckpointing()
//    env.getCheckpointConfig.setCheckpointingMode()


    val path: String = AdStatisticsByGeo.getClass.getClassLoader.getResource("AdClickLog.csv").getPath

//    env.addSource()

    val value = env.readTextFile(path)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          AdClickLog(dataArrays(0).toLong, dataArrays(1).toLong, dataArrays(2), dataArrays(3), dataArrays(4).toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[AdClickLog](Time.seconds(10)) {
        override def extractTimestamp(element: AdClickLog): Long = element.timestamp * 1000L
      })


    val keyByData = value.keyBy(data => (data.userId, data.adId)) //根据用户id和用户点击行为做分组
      .process(new MyKeyByProcess(100))


    val filterData = keyByData.keyBy(_.province) //按照省份划分的点击量
      .timeWindow(Time.hours(1), Time.seconds(5))
      .aggregate(new MyAggFunction(), new MyWindowProcess())


    filterData.print("output").setParallelism(1)
    keyByData.getSideOutput(new OutputTag[Warnning]("warnning")).print("warning").setParallelism(1)


    env.execute("AdStatisticsByGeo job")
  }
}

case class AdClickLog(
                       userId: Long,
                       adId: Long,
                       province: String,
                       city: String,
                       timestamp: Long
                     )

case class CountByProvince(
                            province: String,
                            windowEnd: Long,
                            count: Long
                          )

case class Warnning(
                     userId: Long,
                     adId: Long,
                     msg: String
                   )

class MyKeyByProcess(maxCount: Int) extends KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog] {

  lazy val valueState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("value", classOf[Long]))

  lazy val booleanState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("boolean", classOf[Boolean]))

  override def processElement(value: AdClickLog, ctx: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#Context, out: Collector[AdClickLog]): Unit = {
    val count: Long = valueState.value()


    if (count == 0) {
      //获取当前系统时间设置清除第二天状态的定时器
      val ts = (ctx.timerService().currentProcessingTime() / (24 * 60 * 60 * 1000L) + 1) * (24 * 60 * 60 * 1000L)
      ctx.timerService().registerProcessingTimeTimer(ts)

    }

    if (count >= maxCount) {
      val flag: Boolean = booleanState.value()
      if (!flag) {
        booleanState.update(true)
        ctx.output(new OutputTag[Warnning]("warnning"), Warnning(ctx.getCurrentKey._1, ctx.getCurrentKey._2, " click more then 100 !"))
      }
      return
    }

    valueState.update(count + 1)
    out.collect(value)

  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#OnTimerContext, out: Collector[AdClickLog]): Unit = {
    valueState.clear()
    booleanState.clear()
  }
}

class MyAggFunction() extends AggregateFunction[AdClickLog, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(value: AdClickLog, accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

class MyWindowProcess() extends WindowFunction[Long, CountByProvince, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[CountByProvince]): Unit = {
    out.collect(CountByProvince(key, window.getEnd, input.last))
  }
}